package net.bwie.vehicle.dws.job;


import com.alibaba.fastjson.JSON;
import net.bwie.realtime.guanjuntao.util.DateTimeUtil;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import net.bwie.vehicle.dws.bean.CarBean;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author: FuHe
 * @Date: 2025/5/27
 */
public class TryDataAbnormalJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "car-vehicle-data");
//        kafkaStream.print();

        DataStream<String> alarmStream = processheadel(kafkaStream);
        alarmStream.print();
        JdbcUtil.sinkToClickhouseUpsert(alarmStream,
                "INSERT INTO new_car.car_Try (window_start, window_end, vin, alarm_type)\n" +
                        "VALUES (?, ?, ?, ?);");

        env.execute("TryDataAbnormalJob");
    }

    private static DataStream<String> processheadel(DataStream<String> kafkaStream) {
        // 最低电池电量
        Double MinBatteryLevel = 10.0;
        // 最高电池温度
        Double MaxBatteryTemp = 60.0;
        // 最高电机温度
        Double maxMotorTemp = 60.0;
        // 最高速度
        Double maxSpeed = 120.0;
        // 将Kafka流中的消息转换为CarBean对象
        SingleOutputStreamOperator<CarBean> map = kafkaStream.map(o -> JSON.parseObject(o, CarBean.class));

// 为CarBean对象分配时间戳和水印
// 使用有限的无序性策略，允许的最大乱序时间为0秒，即要求数据完全有序
        SingleOutputStreamOperator<CarBean> vehicleDataSingleOutputStreamOperator = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<CarBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<CarBean>() {
                            @Override
                            public long extractTimestamp(CarBean element, long recordTimestamp) {
                                // 从CarBean对象中提取时间戳
                                return element.getTimestamp();
                            }
                        }
                )
        );


        KeyedStream<CarBean, CarBean> vehicleDataVehicleDataKeyedStream = vehicleDataSingleOutputStreamOperator.keyBy(o -> o);
        WindowedStream<CarBean, CarBean, TimeWindow> window = vehicleDataVehicleDataKeyedStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );
        SingleOutputStreamOperator<String> apply = window.apply(
                new WindowFunction<CarBean, String, CarBean, TimeWindow>() {
                    @Override
                    public void apply(CarBean vehicleData, TimeWindow window, Iterable<CarBean> input, Collector<String> out) throws Exception {
                        // 将窗口开始时间从长整型转换为字符串类型，使用指定的日期时间格式
                        String window_start = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
                        // 将窗口结束时间从长整型转换为字符串类型，使用指定的日期时间格式
                        String window_end = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);

                        // 获取车辆数据中的电池电量
                        Double batteryLevel = vehicleData.getBatteryLevel();
                        // 获取车辆数据中的电池温度
                        Double batteryTemp = vehicleData.getBatteryTemp();
                        // 获取车辆数据中的电机温度
                        Double motorTemp = vehicleData.getMotorTemp();
                        // 获取车辆数据中的速度
                        Double speed = vehicleData.getSpeed();
                        // 获取车辆数据中的车辆识别号
                        String vin = vehicleData.getVin();

                        // 电池电量报警
                        if (batteryLevel < MinBatteryLevel) {
                            out.collect(window_start + "," + window_end + "," + vin + "," + " 电池电量低于10%");
                        }
                        // 电池温度报警
                        if (batteryTemp > MaxBatteryTemp) {
                            out.collect(window_start + "," + window_end + "," + vin + "," + " 电池温度过高");
                        }
                        // 电机温度报警
                        if (motorTemp > maxMotorTemp) {
                            out.collect(window_start + "," + window_end + "," + vin + "," + " 电机温度过高");
                        }
                        // 超速报警
                        if (speed > maxSpeed) {
                            out.collect(window_start + "," + window_end + "," + vin + "," + " 超过最高速度");
                        }

                    }
                }
        );
        return apply;
    }


}
